package com.lvmama.rhino.analyze.processing

import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel
import org.elasticsearch.spark.sql._


/**
 * Description:自定义漏斗清洗出用户的操作路径，存入es
 * Date:2017/5/24 16:24
 * Created by wxxuyuan
 */
object WirelessCustomizeFunnel {
  def process(sc: SparkContext, pageForward: DataFrame,date:String): Unit = {
    import com.lvmama.rhino.analyze.client.WirelessStat._

    val commons = Seq("deviceToken","sessionId", "channelCode", "platformCode","pageTypeCode","timestamp", "pageParam","pageCode")

    //获取categoryId
    val getCategoryId = pageForward.select(commons.map(col): _*)
      .withColumn("category_id", when(col("pageParam").getItem("categoryId").isNull, "-1")
                                .otherwise(col("pageParam").getItem("categoryId")))
      .coalesce(20)


    //访问事件
    val pagePV = getCategoryId
      .withColumn("product_id", when(col("pageParam").getItem("productId").isNull, "-1").otherwise(col("pageParam").getItem("productId")))
      .withColumn("pageTotalCode", concat(col("pageCode"), col("category_id"), col("product_id")))
      .withColumn("pageTotalCodeLag", dataLag(col("pageTotalCode")))
      .withColumn("deviceToken",when(col("platformCode") === "2g53",col("sessionId")).otherwise(col("deviceToken")))
      .filter(col("deviceToken").isNotNull && (col("deviceToken") =!= "") && (col("deviceToken") =!= "000000000000000"))//过滤掉deviceToken异常值
      .filter(col("pageTotalCodeLag").isNull.or(col("pageTotalCode") !== col("pageTotalCodeLag"))).coalesce(1)
      .withColumn("operate",concat(col("pageTypeCode"),lit(","), col("category_id"),lit(","), col("product_id")))
      .drop(col("pageTotalCode"))
      .drop(col("pageTotalCodeLag"))
      .drop(col("pageParam"))
      .drop(col("pageCode"))

      .persist(StorageLevel.MEMORY_AND_DISK_SER)

//    pagePV.show(50,false)

    val operateDate = date.replaceAll("/","")

    val operatePath = pagePV.groupBy("sessionId","platformCode")
      .agg(concat_ws("->",collect_list("operate")) as "operatePath",
        count("*") as "pathLength")
      .coalesce(20)
        .filter(col("pathLength") <= 256)
      .select(col("sessionId"),col("platformCode"),lit(operateDate) as "operateDate",col("operatePath"))
      .withColumn("sessionId",concat(col("sessionId"),col("operateDate")))

    operatePath.saveToEs("flow_v1/sessionOperPath",Map("es.mapping.id" -> "sessionId"))


    val operatePath_dt = pagePV.groupBy("deviceToken","platformCode")
      .agg(concat_ws("->",collect_list("operate")) as "operatePath",
        count("*") as "pathLength")
      .coalesce(20)
      .filter(col("pathLength") <= 256)
      .select(col("deviceToken"),col("platformCode"),lit(operateDate) as "operateDate",col("operatePath"))
      .withColumn("deviceToken",concat(col("deviceToken"),col("operateDate")))
    operatePath_dt.saveToEs("flow_v1/dtOperPath",Map("es.mapping.id" -> "deviceToken"))

    pagePV.unpersist()
  }
}
